package com.flink.wc.demo.demo_20240801;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Demo_01 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        DataStreamSource<Person> personDataStreamSource = env.fromElements(
//                new Person("zhangsan", 18),
//                new Person("lisi", 19),
//                new Person("wangwu", 20)
//        );
        DataStreamSource<String> personDataStreamSource = env.readTextFile("E:\\mywork\\FlinkDemo-1.19\\src\\main\\java\\com\\flink\\wc\\demo\\demo_20240801\\test.txt");
        SingleOutputStreamOperator<Person> map = personDataStreamSource.map(line -> new Person(line.split(",")[0], new Integer(line.split(",")[1])));
        DataStream<Person> filter = map.filter(person -> person.age > 18);
        filter.print();
//        filter.writeAsText("E:\\mywork\\FlinkDemo-1.19\\src\\main\\java\\com\\flink\\wc\\demo\\demo_20240801\\test_out.txt");
        env.execute();
    }
    public static class Person {
        public String name;
        public Integer age;
        public Person() {}

        public Person(String name, Integer age) {
            this.name = name;
            this.age = age;
        }

        @Override
        public String toString() {
            return this.name.toString() + ": age " + this.age.toString();
        }
    }
}
